package net.boot.smpp.task;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import net.boot.smpp.listener.SmppListener;
import net.boot.smpp.listener.impl.SessionStateListenerImpl;
import org.jsmpp.session.SMPPServerSession;
import org.jsmpp.session.SMPPServerSessionListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author qiusu
 * @since  2025/08/18
 */
@Slf4j
@Component
public class SmppServerTask implements Runnable{

    private final AtomicInteger requestCounter = new AtomicInteger();
    private final AtomicBoolean running = new AtomicBoolean(false);

    private SMPPServerSessionListener sessionListener;

    @Resource
    private SmppPoolTask smppPoolTask;

    @Resource
    private SmppListener smppListener;

    @Value("${jsmpp.server.procDegree:10}")
    private int processorDegree;

    @Value("${jsmpp.server.port:8056}")
    private int port;

    private Thread serverThread;

    @PostConstruct
    public void start() {
        if (running.compareAndSet(false, true)) {
            serverThread = new Thread(this, "SmppListenerThread");
            serverThread.start();
            log.info("SMPP Listener started");
        }
    }

    @PreDestroy
    public void stop() {
        if (running.compareAndSet(true, false)) {
            log.info("Stopping SMPP listener...");

            // 关闭监听器
            if (sessionListener != null) {
                try {
                    sessionListener.close();
                } catch (IOException e) {
                    log.warn("Error closing session listener", e);
                }
            }

            // 中断服务器线程
            if (serverThread != null) {
                serverThread.interrupt();
            }

            log.info("SMPP listener stopped");
        }
    }

    @Override
    public void run() {
        try {
            sessionListener = new SMPPServerSessionListener(port);
            sessionListener.setSessionStateListener(new SessionStateListenerImpl());
            sessionListener.setPduProcessorDegree(processorDegree);
            new TrafficWatcherThread().start();
            log.info("Listening on port {}", port);

            while (running.get() && !Thread.currentThread().isInterrupted()) {
                try {
                    SMPPServerSession serverSession = sessionListener.accept();
                    log.info("Accepting connection for session {}", serverSession.getSessionId());
                    serverSession.setMessageReceiverListener(smppListener);
                    //waitBindExecService.execute(new WaitBindTask(serverSession, 60000));
                    smppPoolTask.waitBindTask(serverSession, 60000);
                } catch (IOException e) {
                    if (running.get() && !Thread.currentThread().isInterrupted()) {
                        log.error("I/O error occurred", e);
                    }
                    // 当running为false或线程被中断时，这是预期的关闭操作，不需要记录错误
                }
            }
        } catch (IOException e) {
            log.error("Failed to start SMPP listener", e);
        } finally {
            // 确保资源被清理
            if (sessionListener != null) {
                try {
                    sessionListener.close();
                } catch (IOException e) {
                    log.warn("Error closing session listener", e);
                }
            }
        }

        log.info("SMPP listener thread finished");
    }

    private class TrafficWatcherThread extends Thread {
        public TrafficWatcherThread() {
            super("TrafficWatcherThread");
        }

        @Override
        public void run() {
            log.info("Starting traffic watcher...");
            while (running.get() && !Thread.currentThread().isInterrupted()) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
                int requestsPerSecond = requestCounter.getAndSet(0);
                if (requestsPerSecond != 0) {
                    log.debug("Requests per second: {}", requestsPerSecond);
                }
            }
            log.info("Traffic watcher stopped");
        }
    }
}
